package log_mqtt

import (
	"fmt"
	"server/global"
	"server/utils"
	"time"

	MQTT "github.com/eclipse/paho.mqtt.golang"
)

var messageHandler MQTT.MessageHandler = func(client MQTT.Client, msg MQTT.Message) {
	// fmt.Println("************************** lock message ****************************")
	// fmt.Printf("topic: %s\n", msg.Topic())
	// fmt.Printf("message: %s\n", msg.Payload())
	// global.LOG.Info("***** LOG *****")
	global.LOG.Debug(fmt.Sprintf("topic: %s", msg.Topic()))
	global.LOG.Debug(fmt.Sprintf("message: %s", msg.Payload()))
}

type Client struct {
	MQTT.Client
	Host     string
	ClientID string
	UserName string
	Password string
}

type Message struct {
	ClientID string `json:"clientId"`
	Type     string `json:"type"`
	Data     string `json:"data,omitempty"`
	Time     int64  `json:"time"`
}

func (m *Client) NewClient() {
	ClientID := "go-log-" + utils.GenerateTimeMD5()
	opts := MQTT.NewClientOptions()
	opts.AddBroker(global.CONFIG.Emqx.Host)
	opts.SetClientID(ClientID)
	opts.SetUsername(global.CONFIG.Emqx.Username)
	opts.SetPassword(global.CONFIG.Emqx.Password)
	opts.SetCleanSession(true)
	opts.SetAutoReconnect(true)
	opts.SetKeepAlive(120 * time.Second)
	opts.SetPingTimeout(10 * time.Second)
	opts.SetWriteTimeout(10 * time.Second)
	opts.SetOnConnectHandler(func(client MQTT.Client) {
		// 连接被建立后的回调函数
		global.LOG.Info(fmt.Sprintf(" Mqtt is connected! clientId: %v", ClientID))
		// 重连后重新订阅
		m.Subscribe()
	})
	opts.SetConnectionLostHandler(func(client MQTT.Client, err error) {
		// 连接被关闭后的回调函数
		global.LOG.Info(fmt.Sprintf("Mqtt is disconnected! clientId: %v reason: %v", ClientID, err.Error()))
	})

	m.Client = MQTT.NewClient(opts)
}

func (m *Client) Connect() {
	global.LOG.Info("connect")

	token := m.Client.Connect()
	if token.Wait() && token.Error() != nil {
		global.LOG.Error(fmt.Sprintf("%v", token.Error()))
	}
}

func (m *Client) DisConnect() {
	fmt.Println("disconnect")
	m.Client.Disconnect(250)
}

func (m *Client) Subscribe() {
	global.LOG.Info("subscribe")
	topics := make(map[string]byte)
	// topics["$SYS/brokers/+/clients/#"] = 0
	for _, v := range global.CONFIG.Emqx.LogTopics {
		topics[v] = 0
	}
	global.LOG.Info(fmt.Sprintf("%v", topics))
	// token := m.Client.Subscribe("dqt201/+/tx", 0, messageHandler)
	token := m.Client.SubscribeMultiple(topics, messageHandler)
	if token.Wait() && token.Error() != nil {
		global.LOG.Error(fmt.Sprintf("%v", token.Error()))
	}
}

func (m *Client) Pub() {
	global.LOG.Info("publish")
	s := time.Now().String()
	token := m.Client.Publish("date", 0, false, s)
	if token.Wait() && token.Error() != nil {
		global.LOG.Error(fmt.Sprintf("%v", token.Error()))
	}
}

func (m *Client) Run() {
	m.NewClient()
	m.Connect()
}
